package com.bbx.flink.demo.evictors;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;

@Slf4j
public class CountEvictorDemo {

    /**
     * Evictor 预定的实现类  CountEvictors
     * @param args
     * @throws Exception
     */
    public  static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.10.131",10003)
                .map((MapFunction<String ,Integer>)value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(1))
                //CountEvictor  从窗口的元素中剔除超出范围的元素，从窗口的源头开始剔除
                //例：窗口内元素为 1，3 ，5 ，7 ，9 ，11   ，指定的剔除值为4   ，则 窗口内只会保留  5，7，9，11
                .evictor(CountEvictor.of(4))
                .sum(0)
                .print();
        env.execute();
    }
}